package com.aiwiown.snackmq.storage.service;

import com.aiwiown.snackmq.storage.config.StorageConfig;
import com.aiwiown.snackmq.storage.file.MappedFileQueue;
import com.aiwiown.snackmq.storage.index.IndexEntry;
import com.aiwiown.snackmq.storage.index.IndexManager;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Slf4j
public class MessageStoreService {
    private final StorageConfig config;
    private MappedFileQueue messageQueue;
    private IndexManager indexManager;
    private final AtomicLong messageCount;
    private final ScheduledExecutorService scheduler;
    // 将 storeLock 改为私有静态常量以确保线程安全
    private static final Object STORE_LOCK = new Object();

    public MessageStoreService(StorageConfig config) {
        this.config = config;
        this.messageCount = new AtomicLong(0);
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
    }

    /**
     * 初始化存储服务
     */
    public void init() {
        try {
            // 创建必要的目录
            createDirectories();

            // 初始化消息队列
            messageQueue = new MappedFileQueue(config);
            messageQueue.init();

            // 初始化索引管理器
            indexManager = new IndexManager(config);
            indexManager.init();

            // 恢复写入位置
            messageQueue.restoreWritePosition();
        } catch (Exception e) {
            log.error("Failed to initialize message store service", e);
            throw new RuntimeException("Failed to initialize message store service", e);
        }
    }

    /**
     * 创建必要的目录
     */
    private void createDirectories() {
        try {
            // 创建消息存储目录
            File messageDir = new File(config.getMessageStorePath());
            if (!messageDir.exists()) {
                messageDir.mkdirs();
            }

            // 创建索引存储目录
            File indexDir = new File(config.getIndexStorePath());
            if (!indexDir.exists()) {
                indexDir.mkdirs();
            }

            log.info("Created directories: messages={}, index={}",
                    messageDir.getAbsolutePath(), indexDir.getAbsolutePath());
        } catch (Exception e) {
            log.error("Failed to create directories", e);
            throw new RuntimeException("Failed to create directories", e);
        }
    }

    /**
     * 启动定时任务
     */
    private void startScheduledTasks() {
        // 定时刷盘
        scheduler.scheduleAtFixedRate(() -> {
            try {
                messageQueue.flushAsync();
            } catch (Exception e) {
                log.error("Error in scheduled flush task", e);
            }
        }, config.getFlushInterval(), config.getFlushInterval(), TimeUnit.MILLISECONDS);

        // 定时清理过期文件
        if (config.isRetainFiles()) {
            scheduler.scheduleAtFixedRate(() -> {
                try {
                    long expireTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(7); // 默认保留7天
                    messageQueue.deleteExpiredFiles(expireTime);
                } catch (Exception e) {
                    log.error("Error in scheduled cleanup task", e);
                }
            }, 1, 1, TimeUnit.HOURS);
        }
    }

    /**
     * 存储消息
     */
    public String storeMessage(byte[] message) {
        if (message == null || message.length == 0) {
            log.error("Invalid message: null or empty");
            return null;
        }

        // Validate message length against maximum allowed size
        long mappedFileSize = config.getMappedFileSize();
        long maxAllowedLength = mappedFileSize - 1024;
        if (message.length > maxAllowedLength) {
            log.error("Message too large: length={}, maxAllowed={}", message.length, maxAllowedLength);
            return null;
        }
        // 生成消息ID（线程安全操作）
        String messageId = UUID.randomUUID().toString();
        // 创建带长度前缀的消息（线程安全操作）
        byte[] messageWithPrefix = new byte[message.length + 4];
        ByteBuffer buffer = ByteBuffer.wrap(messageWithPrefix);
        buffer.order(ByteOrder.BIG_ENDIAN);  // 使用大端序
        buffer.putInt(message.length);  // 写入消息长度前缀
        System.arraycopy(message, 0, messageWithPrefix, 4, message.length);
        // 确保只有实际需要同步的部分被同步
        try {
            synchronized (STORE_LOCK) {
                // 写入消息（共享资源操作）
                long offset = messageQueue.writeMessage(messageWithPrefix);
                if (offset < 0) {
                    log.error("Failed to write message: {}", messageId);
                    return null;
                }
                // 获取当前文件索引
                int fileIndex = messageQueue.getMappedFileCount() - 1;
                if (fileIndex < 0) {
                    fileIndex = 0;
                }
                // 添加索引
                long storeTime = System.currentTimeMillis();
                IndexEntry indexEntry = new IndexEntry(messageId, fileIndex, offset, messageWithPrefix.length, storeTime);
                indexManager.addIndex(indexEntry);

                // 更新消息计数
                messageCount.incrementAndGet();
                return messageId;
            }
        } catch (Exception e) {
            log.error("Error storing message", e);
            return null;
        }
    }

    /**
     * 获取消息
     */
    public byte[] getMessage(String messageId) {
        try {
            // 获取消息索引
            IndexEntry index = indexManager.getIndex(messageId);
            if (index == null) {
                log.warn("No index found for message: {}", messageId);
                return null;
            }
            // 从文件读取消息（包含长度前缀）
            byte[] messageWithPrefix = messageQueue.readMessage(index.getOffset(), index.getPosition());
            
            // 检查读取是否成功
            if (messageWithPrefix == null) {
                log.error("Failed to read message from storage: messageId={}, offset={}, position={}", 
                         messageId, index.getOffset(), index.getPosition());
                return null;
            }

            // 读取长度前缀（使用大端序）
            ByteBuffer buffer = ByteBuffer.wrap(messageWithPrefix, 0, 4);
            buffer.order(ByteOrder.BIG_ENDIAN);
            int length = buffer.getInt();

            // 增强验证逻辑
            if (length <= 0 || length > config.getMappedFileSize() - 1024) {
                log.error("Invalid message length: length={}, max={}, messageId={}",
                        length, config.getMappedFileSize() - 1024, messageId);
                // 异常消息删除索引
                indexManager.removeIndex(messageId);
                return null;
            }

            if (length != messageWithPrefix.length - 4) {
                log.error("Invalid message length: expected={}, actual={}, total length={}",
                        length, messageWithPrefix.length - 4, messageWithPrefix.length);
                // 异常消息删除索引
                indexManager.removeIndex(messageId);
                return null;
            }
            // 返回实际消息内容（不包含长度前缀）
            byte[] message = new byte[length];
            System.arraycopy(messageWithPrefix, 4, message, 0, length);
            return message;
        } catch (Exception e) {
            log.error("Error retrieving message: {}", messageId, e);
            return null;
        }
    }

    /**
     * 删除消息
     */
    public boolean deleteMessage(String messageId) {
        try {
            IndexEntry indexEntry = indexManager.getIndex(messageId);
            if (indexEntry == null) {
                return false;
            }

            // 删除索引
            indexManager.removeIndex(messageId);
            // 更新消息计数
            messageCount.decrementAndGet();
            return true;
        } catch (Exception e) {
            log.error("Failed to delete message: {}", messageId, e);
            return false;
        }
    }

    /**
     * 获取时间范围内的消息
     */
    public List<byte[]> getMessagesByTimeRange(long startTime, long endTime) throws IOException {
        List<IndexEntry> indexesByTimeRange = indexManager.getIndexesByTimeRange(startTime, endTime);
        List<byte[]> list = new ArrayList<>();
        for (IndexEntry entry : indexesByTimeRange) {
            byte[] message = messageQueue.readMessage(entry.getStartPosition(), entry.getPosition());
            if (message != null && message.length == entry.getPosition()) {
                list.add(message);
            } else {
                log.error("Failed to read message at offset={}, length={}",
                        entry.getStartPosition(), entry.getPosition());
            }
        }
        return list;
    }

    /**
     * 获取消息数量
     */
    public long getMessageCount() {
        return messageCount.get();
    }

    /**
     * 获取存储大小
     */
    public long getStoreSize() {
        return messageQueue.getTotalSize();
    }

    /**
     * 获取索引数量
     */
    public long getIndexCount() {
        return indexManager.getTotalIndexCount();
    }

    /**
     * 批量存储消息
     */
    public List<String> storeMessages(List<byte[]> messages) {
        return messages.stream()
                .map(this::storeMessage)
                .filter(messageId -> messageId != null)
                .collect(Collectors.toList());
    }

    /**
     * 批量获取消息
     */
    public List<byte[]> getMessages(List<String> messageIds) {
        if (messageIds == null || messageIds.isEmpty()) {
            log.warn("Message IDs list is null or empty");
            return Collections.emptyList();
        }

        List<byte[]> messages = new ArrayList<>();
        List<String> failedIds = new ArrayList<>();

        for (String messageId : messageIds) {
            try {
                byte[] message = getMessage(messageId);
                if (message != null) {
                    messages.add(message);
                } else {
                    failedIds.add(messageId);
                    log.error("Failed to read message: {}", messageId);
                }
            } catch (Exception e) {
                failedIds.add(messageId);
                log.error("Error reading message: {}", messageId, e);
            }
        }

        if (!failedIds.isEmpty()) {
            log.error("Failed to read messages: {}", failedIds);
        }

        return messages;
    }

    /**
     * 批量删除消息
     */
    public boolean deleteMessages(List<String> messageIds) {
        try {
            for (String messageId : messageIds) {
                if (!deleteMessage(messageId)) {
                    return false;
                }
            }
            return true;
        } catch (Exception e) {
            log.error("Failed to delete messages", e);
            return false;
        }
    }

    /**
     * 获取指定时间范围内的消息ID列表
     */
    public List<String> getMessageIdsByTimeRange(long startTime, long endTime) {
        return indexManager.getIndexesByTimeRange(startTime, endTime).stream()
                .map(IndexEntry::getMessageId)
                .collect(Collectors.toList());
    }

    /**
     * 获取指定文件索引的消息ID列表
     */
    public List<String> getMessageIdsByFileIndex(int fileIndex) {
        return indexManager.getAllIndexes().values().stream()
                .filter(entry -> entry.getFileIndex() == fileIndex)
                .map(IndexEntry::getMessageId)
                .collect(Collectors.toList());
    }

    /**
     * 获取指定大小范围内的消息
     */
    public List<byte[]> getMessagesBySizeRange(int minSize, int maxSize) {
        List<byte[]> result = new ArrayList<>();
        try {
            // 获取所有消息的索引
            Map<String, IndexEntry> allIndexes = indexManager.getAllIndexes();
            // 遍历所有索引
            for (IndexEntry index : allIndexes.values()) {
                // 考虑4字节的长度前缀
                int actualLength = index.getPosition() - 4;
                // 检查消息长度是否在范围内
                if (actualLength >= minSize && actualLength <= maxSize) {
                    byte[] message = getMessage(index.getMessageId());
                    if (message != null) {
                        result.add(message);
                    }
                }
            }
        } catch (Exception e) {
            log.error("Error querying messages by size range", e);
        }
        return result;
    }

    /**
     * 清理过期消息
     */
    public void cleanupExpiredMessages(long expireTime) {
        List<String> expiredMessageIds = indexManager.getIndexesByTimeRange(0, expireTime).stream()
                .map(IndexEntry::getMessageId)
                .collect(Collectors.toList());

        if (!expiredMessageIds.isEmpty()) {
            deleteMessages(expiredMessageIds.stream().map(String::valueOf).collect(Collectors.toList()));
            log.info("Cleaned up {} expired messages", expiredMessageIds.size());
        }
    }

    /**
     * 强制刷盘
     */
    public void flush() {
        try {
            messageQueue.flush();
        } catch (Exception e) {
            log.error("Failed to flush message store", e);
        }
    }

    /**
     * 获取映射文件数量
     */
    public int getMappedFileCount() {
        return messageQueue.getMappedFileCount();
    }

    /**
     * 获取存储统计信息
     */
    public StorageStats getStorageStats() {
        return new StorageStats(
                messageCount.get(),
                messageQueue.getTotalSize(),
                indexManager.getTotalIndexCount(),
                messageQueue.getMappedFileCount(),
                indexManager.getIndexFileCount()
        );
    }

    /**
     * 存储统计信息
     */
    @lombok.Data
    public static class StorageStats {
        private final long messageCount;
        private final long storeSize;
        private final long indexCount;
        private final int mappedFileCount;
        private final int indexFileCount;
    }

    /**
     * 关闭存储服务
     */
    public void shutdown() {
        try {
            log.info("开始关闭 MessageStoreService...");
            
            // 关闭定时任务
            if (scheduler != null && !scheduler.isShutdown()) {
                log.debug("关闭定时任务调度器...");
                scheduler.shutdown();
                try {
                    if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                        log.warn("定时任务调度器未能在5秒内关闭，强制关闭");
                        scheduler.shutdownNow();
                    }
                } catch (InterruptedException e) {
                    log.warn("等待定时任务关闭时被中断", e);
                    scheduler.shutdownNow();
                    Thread.currentThread().interrupt();
                }
            }

            // 强制刷盘
            flush();

            // 保存写入位置
            if (messageQueue != null) {
                messageQueue.saveWritePosition();
            }

            // 关闭资源 - 使用 ResourceManager 确保可靠关闭
            if (messageQueue != null) {
                log.debug("关闭消息队列...");
                messageQueue.close();
                messageQueue = null;
            }
            
            if (indexManager != null) {
                log.debug("关闭索引管理器...");
                indexManager.close();
                indexManager = null;
            }

            log.info("MessageStoreService 关闭完成");
        } catch (Exception e) {
            log.error("Error during shutdown", e);
        }
    }

    /**
     * 删除所有文件
     */
    public void delete() {
        try {
            log.info("开始删除 MessageStoreService 所有文件...");
            
            // 先关闭服务
            shutdown();
            
            // 删除消息队列文件
            if (messageQueue != null) {
                log.debug("删除消息队列文件...");
                messageQueue.delete();
            }

            // 删除索引管理器文件
            if (indexManager != null) {
                log.debug("删除索引管理器文件...");
                indexManager.delete();
            }

            log.info("MessageStoreService 文件删除完成");
        } catch (Exception e) {
            log.error("Error during delete", e);
        }
    }
}